import { listenFromTopic, Options } from "@nodejs/mq"
import { responseFromBuffer } from "./proto"
import { isErrorMessage, isResponseMessage, isSendToMessage, isSendToRoomMessage } from "@core/types"
import { WebSocket } from "ws"
import { parseJSON } from "@tools/core"
import dayjs from "dayjs"

export function listenWsResponseQueue(uuid: string, rootProto: protobuf.Root, ws: WebSocket, options: Options.AssertQueue = {
    exclusive: true
}, cid: string = '') {
    if (cid && cid.length > 0) {
        return listenFromTopic(`response.topic`, `response.queue.${uuid}.${cid}`, `response.ws`, async (msg, channel) => {
            const response = responseFromBuffer(rootProto, msg.content)
            if (isErrorMessage(response)) {
                const payload = response.payload
                if (ws.readyState === WebSocket.OPEN) {
                    ws.send(JSON.stringify({ type: response.type, message: payload.message, stack: payload.stack, data: parseJSON(payload.data.toString()) }))
                    channel.ack(msg)
                    return;
                }
                channel.nack(msg)
                return;
            }
            if (isSendToMessage(response)) {
                const payload = response.payload
                if (ws.readyState === WebSocket.OPEN && payload.to === uuid) {
                    ws.send(payload.data.toString())
                    channel.ack(msg)
                    return;
                }
                channel.nack(msg)
                return;
            }
            // 群发
            if (isSendToRoomMessage(response)) {
                const payload = response.payload
                if (ws.readyState === WebSocket.OPEN) {
                    // 判断是否已经发送过了
                    ws.send(payload.data.toString())
                    channel.ack(msg)
                    return;
                }
                channel.nack(msg)
                return;
            }
            if (isResponseMessage(response)) {
                const payload = response.payload
                if (ws.readyState === WebSocket.OPEN) {
                    ws.send(payload.data.toString())
                    channel.ack(msg)
                    return;
                }
                channel.nack(msg)
                return;
            }
        }, options)
    }
    return listenFromTopic(`response.topic`, `response.queue.${uuid}`, `response.${uuid}`, async (msg, channel) => {
        const response = responseFromBuffer(rootProto, msg.content)
        if (isErrorMessage(response)) {
            const payload = response.payload
            ws.send(JSON.stringify({ type: response.type, message: payload.message, stack: payload.stack, data: parseJSON(payload.data.toString()) }))
            console.log(`ws response is error, finish a request ${dayjs().format('YYYY-MM-DD HH:mm:ss')}`)
            channel.ack(msg)
            return;
        }
        if (isSendToMessage(response)) {
            const payload = response.payload
            ws.send(payload.data.toString())
            console.log(`ws response is send to, finish a request ${dayjs().format('YYYY-MM-DD HH:mm:ss')}`)
            channel.ack(msg)
            return;
        }
        // 群发
        if (isSendToRoomMessage(response)) {
            const payload = response.payload
            // 判断是否已经发送过了
            ws.send(payload.data.toString())
            console.log(`ws response is send to room, finish a request ${dayjs().format('YYYY-MM-DD HH:mm:ss')}`)
            channel.ack(msg)
            return;
        }
        if (isResponseMessage(response)) {
            const payload = response.payload
            ws.send(payload.data.toString())
            console.log(`ws response is response, finish a request ${dayjs().format('YYYY-MM-DD HH:mm:ss')}`)
            channel.ack(msg)
            return;
        }
    }, options)
}